import java.io.IOException;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
import java.util.UUID;

import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class MensagemFactory {

	private static byte[] TABELA_MSG = Bytes.toBytes("mensagem");
	private static byte[] ATRIBUTOS_MSG = Bytes.toBytes("atributos");
	private static byte[] TEXT = Bytes.toBytes("text");

	private static final int ID_LENGTH = UUID.randomUUID().toString().length();

	public void postarMensagem(String userId, String text) throws IOException {

		if (userId.length() != ID_LENGTH) {
			throw new IOException("UserId is not valid!");
		}

		HTable htable = new HTable(TABELA_MSG);
		Put put = new Put(createId(userId));
		put.add(ATRIBUTOS_MSG, TEXT, Bytes.toBytes(text));
		htable.put(put);
	}

	public List<String> getMensagens(String userId) throws IOException {
		List<String> texts = new LinkedList<String>();
		HTable htable = new HTable(TABELA_MSG);
		Scan scan = new Scan();
		scan.addColumn(ATRIBUTOS_MSG, TEXT);
		scan.setStartRow(createFirstRow(userId));
		scan.setStopRow(createLastRow(userId));
		ResultScanner scanner = htable.getScanner(scan);
		Result rowResult = null;
		while ((rowResult = scanner.next()) != null){
			byte [] value = rowResult.getValue(ATRIBUTOS_MSG, TEXT);
			texts.add(Bytes.toString(value));
		}
		return texts;
	}

	private byte[] createId(String userId) {
		long time = -1 * System.currentTimeMillis();
		return Bytes.add(Bytes.toBytes(userId), Bytes.toBytes(time));
	}
	
	private byte[] createFirstRow(String userId) {
		byte [] ZEROS = new byte [Bytes.SIZEOF_LONG];
		Arrays.fill(ZEROS, (byte)0x0);
		return Bytes.add(Bytes.toBytes(userId), ZEROS);
	}
	
	private byte[] createLastRow(String userId) {
		byte [] ONES = new byte [Bytes.SIZEOF_LONG];
		Arrays.fill(ONES, (byte)0xFF);
		return Bytes.add(Bytes.toBytes(userId), ONES);
	}

	public static void main(String[] args) throws IOException {
		MensagemFactory mensagemFactory = new MensagemFactory();
		mensagemFactory.postarMensagem(args[0], args[1]);
		System.out.println(mensagemFactory.getMensagens(args[0]));
	}
}